{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup\n",
    "\n",
    "**NOTE:** Before running this notebook, be sure to set the stack name in the first code cell to match the name of the CloudFormation stack you used to create this notebook instance. If you used the default stack name, you should not need to make any updates.\n",
    "\n",
    "This notebook performs the following setup actions for this example use of Amazon SageMaker Feature Store:\n",
    "\n",
    "1. Create online-only feature groups\n",
    "2. Create an Amazon Kinesis data stream\n",
    "3. Create an Amazon Kinesis Data Applications (KDA) application"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get ARN's of Lambda functions from CloudFormation stack outputs\n",
    "1. InvokeFraudEndpointLambdaARN\n",
    "2. StreamingAggLambdaARN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing /home/ec2-user/.cache/pip/wheels/3f/5f/a2/ae75ce6341001c1605ccc4675808a04f5eb1d0b441ffec4b88/sagemaker-2.19.0-py2.py3-none-any.whl\n",
      "Requirement already satisfied: protobuf3-to-dict>=0.1.5 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (0.1.5)\n",
      "Requirement already satisfied: importlib-metadata>=1.4.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (1.5.0)\n",
      "Collecting boto3>=1.16.32\n",
      "  Downloading boto3-1.16.33.tar.gz (97 kB)\n",
      "\u001b[K     |████████████████████████████████| 97 kB 9.0 MB/s  eta 0:00:01\n",
      "\u001b[?25hRequirement already satisfied: attrs in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (19.3.0)\n",
      "Requirement already satisfied: numpy>=1.9.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (1.18.1)\n",
      "Collecting smdebug-rulesconfig>=1.0.0\n",
      "  Using cached smdebug_rulesconfig-1.0.0-py3-none-any.whl (14 kB)\n",
      "Requirement already satisfied: google-pasta in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (0.2.0)\n",
      "Requirement already satisfied: packaging>=20.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (20.1)\n",
      "Requirement already satisfied: protobuf>=3.1 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from sagemaker==2.19.0) (3.11.4)\n",
      "Requirement already satisfied: six in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from protobuf3-to-dict>=0.1.5->sagemaker==2.19.0) (1.14.0)\n",
      "Requirement already satisfied: zipp>=0.5 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from importlib-metadata>=1.4.0->sagemaker==2.19.0) (2.2.0)\n",
      "Collecting botocore<1.20.0,>=1.19.33\n",
      "  Using cached botocore-1.19.33-py2.py3-none-any.whl (7.0 MB)\n",
      "Requirement already satisfied: jmespath<1.0.0,>=0.7.1 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from boto3>=1.16.32->sagemaker==2.19.0) (0.10.0)\n",
      "Requirement already satisfied: s3transfer<0.4.0,>=0.3.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from boto3>=1.16.32->sagemaker==2.19.0) (0.3.3)\n",
      "Requirement already satisfied: pyparsing>=2.0.2 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from packaging>=20.0->sagemaker==2.19.0) (2.4.6)\n",
      "Requirement already satisfied: setuptools in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from protobuf>=3.1->sagemaker==2.19.0) (45.2.0.post20200210)\n",
      "Requirement already satisfied: python-dateutil<3.0.0,>=2.1 in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from botocore<1.20.0,>=1.19.33->boto3>=1.16.32->sagemaker==2.19.0) (2.8.1)\n",
      "Requirement already satisfied: urllib3<1.27,>=1.25.4; python_version != \"3.4\" in /home/ec2-user/anaconda3/envs/python3/lib/python3.6/site-packages (from botocore<1.20.0,>=1.19.33->boto3>=1.16.32->sagemaker==2.19.0) (1.25.10)\n",
      "Building wheels for collected packages: boto3\n",
      "  Building wheel for boto3 (setup.py) ... \u001b[?25ldone\n",
      "\u001b[?25h  Created wheel for boto3: filename=boto3-1.16.33-py2.py3-none-any.whl size=128452 sha256=2a67d9c7b00e5be6467738f9050225988c3d5051bf683378ce77fffacf5f3c40\n",
      "  Stored in directory: /home/ec2-user/.cache/pip/wheels/8d/78/40/33932039a5c4c45176211822d559f2c6d20f5ef1cbbce39aa3\n",
      "Successfully built boto3\n",
      "\u001b[31mERROR: awscli 1.18.169 has requirement botocore==1.19.9, but you'll have botocore 1.19.33 which is incompatible.\u001b[0m\n",
      "Installing collected packages: botocore, boto3, smdebug-rulesconfig, sagemaker\n",
      "  Attempting uninstall: botocore\n",
      "    Found existing installation: botocore 1.19.9\n",
      "    Uninstalling botocore-1.19.9:\n",
      "      Successfully uninstalled botocore-1.19.9\n",
      "  Attempting uninstall: boto3\n",
      "    Found existing installation: boto3 1.16.9\n",
      "    Uninstalling boto3-1.16.9:\n",
      "      Successfully uninstalled boto3-1.16.9\n",
      "  Attempting uninstall: smdebug-rulesconfig\n",
      "    Found existing installation: smdebug-rulesconfig 0.1.5\n",
      "    Uninstalling smdebug-rulesconfig-0.1.5:\n",
      "      Successfully uninstalled smdebug-rulesconfig-0.1.5\n",
      "  Attempting uninstall: sagemaker\n",
      "    Found existing installation: sagemaker 2.16.1\n",
      "    Uninstalling sagemaker-2.16.1:\n",
      "      Successfully uninstalled sagemaker-2.16.1\n",
      "Successfully installed boto3-1.16.33 botocore-1.19.33 sagemaker-2.19.0 smdebug-rulesconfig-1.0.0\n",
      "\u001b[33mWARNING: You are using pip version 20.0.2; however, version 20.3.1 is available.\n",
      "You should consider upgrading via the '/home/ec2-user/anaconda3/envs/python3/bin/python -m pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install sagemaker==2.19.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Stored 'STACK_NAME' (str)\n"
     ]
    }
   ],
   "source": [
    "STACK_NAME = 'sm-fs-streaming-agg-stack' # if you're not using the default stack name, replace this\n",
    "%store STACK_NAME"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import boto3\n",
    "\n",
    "cf_client = boto3.client('cloudformation')\n",
    "\n",
    "try:\n",
    "    outputs = cf_client.describe_stacks(StackName=STACK_NAME)['Stacks'][0]['Outputs']\n",
    "    for o in outputs:\n",
    "        if o['OutputKey'] == 'IngestLambdaFunctionARN':\n",
    "            lambda_to_fs_arn = o['OutputValue']\n",
    "        if o['OutputKey'] == 'PredictLambdaFunctionARN':\n",
    "            lambda_to_model_arn = o['OutputValue']\n",
    "        if o['OutputKey'] == 'PredictLambdaFunctionName':\n",
    "            predict_lambda_name = o['OutputValue']\n",
    "\n",
    "except:\n",
    "    msg = f'CloudFormation stack {STACK_NAME} was not found. Please set the STACK_NAME properly and re-run this cell'\n",
    "    sys.exit(ValueError(msg))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lambda_to_model_arn: arn:aws:lambda:us-east-1:835319576252:function:InvokeFraudEndpointLambda\n",
      "lambda_to_fs_arn: arn:aws:lambda:us-east-1:835319576252:function:StreamingIngestAggFeatures\n",
      "predict_lambda_name: InvokeFraudEndpointLambda\n"
     ]
    }
   ],
   "source": [
    "print(f'lambda_to_model_arn: {lambda_to_model_arn}')\n",
    "print(f'lambda_to_fs_arn: {lambda_to_fs_arn}')\n",
    "print(f'predict_lambda_name: {predict_lambda_name}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Stored 'lambda_to_model_arn' (str)\n"
     ]
    }
   ],
   "source": [
    "%store lambda_to_model_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Stored 'predict_lambda_name' (str)\n"
     ]
    }
   ],
   "source": [
    "%store predict_lambda_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# to get the latest sagemaker python sdk\n",
    "#!pip install -U sagemaker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import display_html\n",
    "def restartkernel() :\n",
    "    display_html(\"<script>Jupyter.notebook.kernel.restart()</script>\",raw=True)\n",
    "#restartkernel()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports and other setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker import get_execution_role\n",
    "import sagemaker\n",
    "import boto3\n",
    "import json\n",
    "\n",
    "role = get_execution_role()\n",
    "sm = boto3.Session().client(service_name='sagemaker')\n",
    "smfs_runtime = boto3.Session().client(service_name='sagemaker-featurestore-runtime')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create online-only feature groups\n",
    "When using Amazon SageMaker Feature Store, a core design decision is the definition of feature groups. For our credit card fraud detection use case, we have decided to use two of them:\n",
    "\n",
    "1. `cc-agg-fg` - holds aggregate features that will be updated in near real-time (streaming ingestion)\n",
    "2. `cc-agg-batch-fg` - holds aggregate features that will be updated in batch\n",
    "\n",
    "Establishing a feature group is a one-time step and is done using the `CreateFeatureGroup` API. \n",
    "\n",
    "Feature groups can be created as **online-only**, **offline-only**, or both **online and offline**, which replicates updates from an online store to an offline store in Amazon S3. Since our focus in this example is on demonstrating the use of the feature store for online inference and streaming aggregation of features, we make each of our feature groups online-only.\n",
    "\n",
    "In addition to a feature group name, we provide metadata about each feature in the group. We are using a json file to define the schema, but this is not a requirement. We use a schema file to demonstrate how you might capture the feature group definitions, enabling you to recreate them consistently as you move from a development environment to a test or production environment. In our schema file, we also highlight the record identifier and the event timestamp. All feature groups must have these two features, but you get to decide how to name them.\n",
    "\n",
    "Here is a visual summary of the feature groups we will create below.\n",
    "\n",
    "<img src=\"./images/feature_groups.png\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### cc-agg-fg schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\r\n",
      "    \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated features for each credit card, batch ingestion nightly\"\u001b[39;49;00m,\r\n",
      "    \u001b[94m\"features\"\u001b[39;49;00m: [\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"cc_num\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"bigint\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Credit Card Number (Unique)\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"num_trans_last_10m\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"bigint\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated Metric: Average number of transactions for the card aggregated by past 10 minutes\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"avg_amt_last_10m\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"double\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated Metric: Average transaction amount for the card aggregated by past 10 minutes\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"trans_time\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"double\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Required feature for event timestamp\"\u001b[39;49;00m\r\n",
      "          }\r\n",
      "      ],\r\n",
      "    \r\n",
      "      \u001b[94m\"record_identifier_feature_name\"\u001b[39;49;00m: \u001b[33m\"cc_num\"\u001b[39;49;00m,\r\n",
      "      \u001b[94m\"event_time_feature_name\"\u001b[39;49;00m: \u001b[33m\"trans_time\"\u001b[39;49;00m,\r\n",
      "      \u001b[94m\"tags\"\u001b[39;49;00m: [{\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"Environment\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m : \u001b[33m\"DEV\"\u001b[39;49;00m}, \r\n",
      "               {\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"IngestionType\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m: \u001b[33m\"Streaming\"\u001b[39;49;00m},\r\n",
      "               {\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"CostCenter\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m: \u001b[33m\"C20\"\u001b[39;49;00m}]\r\n",
      "}\r\n"
     ]
    }
   ],
   "source": [
    "!pygmentize schema/cc-agg-fg-schema.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### cc-agg-batch-fg schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\r\n",
      "    \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated features for each credit card, streamed intraday\"\u001b[39;49;00m,\r\n",
      "    \r\n",
      "    \u001b[94m\"features\"\u001b[39;49;00m: [\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"cc_num\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"bigint\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Credit Card Number (Unique)\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"num_trans_last_1w\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"bigint\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated Metric: Average number of transactions for the card aggregated by past 1 week\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"avg_amt_last_1w\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"double\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Aggregated Metric: Average transaction amount for the card aggregated by past 1 week\"\u001b[39;49;00m\r\n",
      "          },\r\n",
      "          {\r\n",
      "              \u001b[94m\"name\"\u001b[39;49;00m: \u001b[33m\"trans_time\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"type\"\u001b[39;49;00m: \u001b[33m\"double\"\u001b[39;49;00m,\r\n",
      "              \u001b[94m\"description\"\u001b[39;49;00m: \u001b[33m\"Required feature for event timestamp\"\u001b[39;49;00m\r\n",
      "          }\r\n",
      "      ],\r\n",
      "    \r\n",
      "      \u001b[94m\"record_identifier_feature_name\"\u001b[39;49;00m: \u001b[33m\"cc_num\"\u001b[39;49;00m,\r\n",
      "      \u001b[94m\"event_time_feature_name\"\u001b[39;49;00m: \u001b[33m\"trans_time\"\u001b[39;49;00m,\r\n",
      "      \u001b[94m\"tags\"\u001b[39;49;00m: [{\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"Environment\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m : \u001b[33m\"DEV\"\u001b[39;49;00m}, \r\n",
      "               {\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"IngestionType\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m: \u001b[33m\"Batch\"\u001b[39;49;00m},\r\n",
      "               {\u001b[94m\"Key\"\u001b[39;49;00m: \u001b[33m\"CostCenter\"\u001b[39;49;00m, \u001b[94m\"Value\"\u001b[39;49;00m: \u001b[33m\"C18\"\u001b[39;49;00m}]\r\n",
      "}\r\n"
     ]
    }
   ],
   "source": [
    "!pygmentize schema/cc-agg-batch-fg-schema.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Utility functions to simplify creation of feature groups\n",
    "`schema_to_defs` takes our schema file and returns feature definitions, and the names of the record identifier and event timestamp feature."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "def schema_to_defs(filename):\n",
    "    schema = json.loads(open(filename).read())\n",
    "    \n",
    "    feature_definitions = []\n",
    "    \n",
    "    for col in schema['Features']:\n",
    "        feature = {'FeatureName': col['name']}\n",
    "        if col['type'] == 'double':\n",
    "            feature['FeatureType'] = 'Fractional'\n",
    "        elif col['type'] == 'bigint':\n",
    "            feature['FeatureType'] = 'Integral'\n",
    "        else:\n",
    "            feature['FeatureType'] = 'String'\n",
    "        feature_definitions.append(feature)\n",
    "\n",
    "    return feature_definitions, schema['record_identifier_feature_name'], schema['event_time_feature_name']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`schema_to_fg` creates a feature group from a schema file. If no s3 URI is passed, an online-only feature group is created."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker import get_execution_role\n",
    "import json\n",
    "\n",
    "def create_feature_group_from_schema(filename, fg_name, role_arn=None, s3_uri=None):\n",
    "    schema = json.loads(open(filename).read())\n",
    "    \n",
    "    feature_defs = []\n",
    "    \n",
    "    for col in schema['features']:\n",
    "        feature = {'FeatureName': col['name']}\n",
    "        if col['type'] == 'double':\n",
    "            feature['FeatureType'] = 'Fractional'\n",
    "        elif col['type'] == 'bigint':\n",
    "            feature['FeatureType'] = 'Integral'\n",
    "        else:\n",
    "            feature['FeatureType'] = 'String'\n",
    "        feature_defs.append(feature)\n",
    "\n",
    "    record_identifier_name = schema['record_identifier_feature_name']\n",
    "    event_time_name = schema['event_time_feature_name']\n",
    "\n",
    "    if role_arn is None:\n",
    "        role_arn = get_execution_role()\n",
    "\n",
    "    if s3_uri is None:\n",
    "        offline_config = {}\n",
    "    else:\n",
    "        offline_config = {'OfflineStoreConfig': {'S3StorageConfig': {'S3Uri': s3_uri}}}\n",
    "        \n",
    "    sm.create_feature_group(\n",
    "        FeatureGroupName = fg_name,\n",
    "        RecordIdentifierFeatureName = record_identifier_name,\n",
    "        EventTimeFeatureName = event_time_name,\n",
    "        FeatureDefinitions = feature_defs,\n",
    "        Description = schema['description'],\n",
    "        Tags = schema['tags'],\n",
    "        OnlineStoreConfig = {'EnableOnlineStore': True},\n",
    "        RoleArn = role_arn,\n",
    "        **offline_config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create the two feature groups"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_feature_group_from_schema('schema/cc-agg-fg-schema.json', 'cc-agg-fg')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_feature_group_from_schema('schema/cc-agg-batch-fg-schema.json', 'cc-agg-batch-fg')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Show that the feature store is aware of the new feature groups"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'FeatureGroupSummaries': [{'FeatureGroupName': 'transaction-feature-group-02-02-21-52',\n",
       "   'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/transaction-feature-group-02-02-21-52',\n",
       "   'CreationTime': datetime.datetime(2020, 12, 2, 2, 21, 54, 95000, tzinfo=tzlocal()),\n",
       "   'FeatureGroupStatus': 'Created',\n",
       "   'OfflineStoreStatus': {'Status': 'Active'}},\n",
       "  {'FeatureGroupName': 'identity-feature-group-02-02-21-52',\n",
       "   'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/identity-feature-group-02-02-21-52',\n",
       "   'CreationTime': datetime.datetime(2020, 12, 2, 2, 21, 52, 666000, tzinfo=tzlocal()),\n",
       "   'FeatureGroupStatus': 'Created',\n",
       "   'OfflineStoreStatus': {'Status': 'Active'}},\n",
       "  {'FeatureGroupName': 'cc-agg-fg',\n",
       "   'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/cc-agg-fg',\n",
       "   'CreationTime': datetime.datetime(2020, 12, 10, 19, 56, 5, 226000, tzinfo=tzlocal()),\n",
       "   'FeatureGroupStatus': 'Creating'},\n",
       "  {'FeatureGroupName': 'cc-agg-batch-fg',\n",
       "   'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/cc-agg-batch-fg',\n",
       "   'CreationTime': datetime.datetime(2020, 12, 10, 19, 56, 8, 580000, tzinfo=tzlocal()),\n",
       "   'FeatureGroupStatus': 'Creating'}],\n",
       " 'ResponseMetadata': {'RequestId': 'e89eab1f-989b-4f56-a651-eb09c82beb0f',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'e89eab1f-989b-4f56-a651-eb09c82beb0f',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '952',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:09 GMT'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sm.list_feature_groups()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Describe each feature group\n",
    "Note that each feature group gets its own ARN, allowing you to manage IAM policies that control access to individual feature groups. The feature names and types are displayed, and the record identifier and event time features are called out specifically. Notice that there is only an `OnlineStoreConfig` and no `OfflineStoreConfig`, as we have decided not to replicate features offline for these groups."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/cc-agg-fg',\n",
       " 'FeatureGroupName': 'cc-agg-fg',\n",
       " 'RecordIdentifierFeatureName': 'cc_num',\n",
       " 'EventTimeFeatureName': 'trans_time',\n",
       " 'FeatureDefinitions': [{'FeatureName': 'cc_num', 'FeatureType': 'Integral'},\n",
       "  {'FeatureName': 'num_trans_last_10m', 'FeatureType': 'Integral'},\n",
       "  {'FeatureName': 'avg_amt_last_10m', 'FeatureType': 'Fractional'},\n",
       "  {'FeatureName': 'trans_time', 'FeatureType': 'Fractional'}],\n",
       " 'CreationTime': datetime.datetime(2020, 12, 10, 19, 56, 5, 226000, tzinfo=tzlocal()),\n",
       " 'OnlineStoreConfig': {'EnableOnlineStore': True},\n",
       " 'RoleArn': 'arn:aws:iam::835319576252:role/service-role/AmazonSageMaker-ExecutionRole-20191006T135881',\n",
       " 'FeatureGroupStatus': 'Creating',\n",
       " 'Description': 'Aggregated features for each credit card, batch ingestion nightly',\n",
       " 'ResponseMetadata': {'RequestId': 'fdf8cf0e-61e3-45b2-8cd0-22c75c1b0433',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'fdf8cf0e-61e3-45b2-8cd0-22c75c1b0433',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '740',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:10 GMT'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sm.describe_feature_group(FeatureGroupName='cc-agg-fg')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'FeatureGroupArn': 'arn:aws:sagemaker:us-east-1:835319576252:feature-group/cc-agg-batch-fg',\n",
       " 'FeatureGroupName': 'cc-agg-batch-fg',\n",
       " 'RecordIdentifierFeatureName': 'cc_num',\n",
       " 'EventTimeFeatureName': 'trans_time',\n",
       " 'FeatureDefinitions': [{'FeatureName': 'cc_num', 'FeatureType': 'Integral'},\n",
       "  {'FeatureName': 'num_trans_last_1w', 'FeatureType': 'Integral'},\n",
       "  {'FeatureName': 'avg_amt_last_1w', 'FeatureType': 'Fractional'},\n",
       "  {'FeatureName': 'trans_time', 'FeatureType': 'Fractional'}],\n",
       " 'CreationTime': datetime.datetime(2020, 12, 10, 19, 56, 8, 580000, tzinfo=tzlocal()),\n",
       " 'OnlineStoreConfig': {'EnableOnlineStore': True},\n",
       " 'RoleArn': 'arn:aws:iam::835319576252:role/service-role/AmazonSageMaker-ExecutionRole-20191006T135881',\n",
       " 'FeatureGroupStatus': 'Creating',\n",
       " 'Description': 'Aggregated features for each credit card, streamed intraday',\n",
       " 'ResponseMetadata': {'RequestId': '619f67b0-c511-41af-bf47-b74847f53407',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': '619f67b0-c511-41af-bf47-b74847f53407',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '743',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:10 GMT'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sm.describe_feature_group(FeatureGroupName='cc-agg-batch-fg')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create an Amazon Kinesis Data Stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "kinesis_client = boto3.client('kinesis')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': 'e7d318c0-396b-9507-bf9f-675c177a49d9',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'e7d318c0-396b-9507-bf9f-675c177a49d9',\n",
       "   'x-amz-id-2': 'xOVfcXh4MRESdOJfAv9cUhJQlNix80/r1IG/TeeavmoJcjp9Tue0KkldV3zpFoFmvla5M/QC0v/qKc43pPqY4O8+r0tDxYvH',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:12 GMT',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '0'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kinesis_client.create_stream(StreamName='cc-stream', ShardCount=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'StreamNames': ['cc-stream', 'dsoaws-kinesis-data-stream'],\n",
       " 'HasMoreStreams': False,\n",
       " 'ResponseMetadata': {'RequestId': 'd9ded67f-2106-145d-8192-a9e20f17c883',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'd9ded67f-2106-145d-8192-a9e20f17c883',\n",
       "   'x-amz-id-2': 'zBikihIQtNE28N+tzudEL4ep7ZxLjt13rja99VGtVDidvuJNQtYcB4WxFUIfP7CM2Q16EASFNrDzPNnOiTIUIurlHLA2hzTE',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:13 GMT',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '81'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kinesis_client.list_streams()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'StreamDescription': {'StreamName': 'cc-stream',\n",
       "  'StreamARN': 'arn:aws:kinesis:us-east-1:835319576252:stream/cc-stream',\n",
       "  'StreamStatus': 'CREATING',\n",
       "  'Shards': [],\n",
       "  'HasMoreShards': False,\n",
       "  'RetentionPeriodHours': 24,\n",
       "  'StreamCreationTimestamp': datetime.datetime(2020, 12, 10, 19, 56, 12, tzinfo=tzlocal()),\n",
       "  'EnhancedMonitoring': [{'ShardLevelMetrics': []}],\n",
       "  'EncryptionType': 'NONE'},\n",
       " 'ResponseMetadata': {'RequestId': 'cd27e41d-5832-969e-956b-9b8076234a40',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'cd27e41d-5832-969e-956b-9b8076234a40',\n",
       "   'x-amz-id-2': 'VoGUqiteRPxViDtXguajhJy6hURCV0BHG4sRuuw7mDGYckVG1PVSZWTilPn5kyfh3kCZBaT1lti0cE9Pw0JY+1Zu+lUIBQfG',\n",
       "   'date': 'Thu, 10 Dec 2020 19:56:13 GMT',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '316'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kinesis_client.describe_stream(StreamName='cc-stream')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting for the Kinesis stream to become active...\n",
      "ACTIVE\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "active_stream = False\n",
    "while not active_stream:\n",
    "    status = kinesis_client.describe_stream(StreamName='cc-stream')['StreamDescription']['StreamStatus']\n",
    "    if (status == 'CREATING'):\n",
    "        print('Waiting for the Kinesis stream to become active...')\n",
    "        time.sleep(20)  \n",
    "    elif (status == 'ACTIVE'): \n",
    "        active_stream = True\n",
    "        print('ACTIVE')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "stream_arn = kinesis_client.describe_stream(StreamName='cc-stream')['StreamDescription']['StreamARN']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'arn:aws:kinesis:us-east-1:835319576252:stream/cc-stream'"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stream_arn"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map the Kinesis stream as an event source for Lambda fraud detection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': 'c8ce7b0e-d352-4ec5-a77f-e7e69486b05d',\n",
       "  'HTTPStatusCode': 202,\n",
       "  'HTTPHeaders': {'date': 'Thu, 10 Dec 2020 19:56:33 GMT',\n",
       "   'content-type': 'application/json',\n",
       "   'content-length': '711',\n",
       "   'connection': 'keep-alive',\n",
       "   'x-amzn-requestid': 'c8ce7b0e-d352-4ec5-a77f-e7e69486b05d'},\n",
       "  'RetryAttempts': 0},\n",
       " 'UUID': '2bf7d8d7-aed9-4b04-ad32-f858e356437f',\n",
       " 'StartingPosition': 'LATEST',\n",
       " 'BatchSize': 100,\n",
       " 'MaximumBatchingWindowInSeconds': 0,\n",
       " 'ParallelizationFactor': 1,\n",
       " 'EventSourceArn': 'arn:aws:kinesis:us-east-1:835319576252:stream/cc-stream',\n",
       " 'FunctionArn': 'arn:aws:lambda:us-east-1:835319576252:function:InvokeFraudEndpointLambda',\n",
       " 'LastModified': datetime.datetime(2020, 12, 10, 19, 56, 33, 787000, tzinfo=tzlocal()),\n",
       " 'LastProcessingResult': 'No records processed',\n",
       " 'State': 'Creating',\n",
       " 'StateTransitionReason': 'User action',\n",
       " 'DestinationConfig': {'OnFailure': {}},\n",
       " 'MaximumRecordAgeInSeconds': 600,\n",
       " 'BisectBatchOnFunctionError': False,\n",
       " 'MaximumRetryAttempts': -1}"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lambda_client = boto3.client('lambda')\n",
    "\n",
    "lambda_client.create_event_source_mapping(EventSourceArn=stream_arn,\n",
    "                                          FunctionName=lambda_to_model_arn,\n",
    "                                          StartingPosition='LATEST',\n",
    "                                          Enabled=True,\n",
    "                                          MaximumRecordAgeInSeconds=60*10\n",
    "                                          ) #DestinationConfig would handle discarded records"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create an Amazon Kinesis Data Applications (KDA) application"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "kda_client = boto3.client('kinesisanalytics')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql_code = 'CREATE OR REPLACE STREAM \"DESTINATION_SQL_STREAM\" (\\n' + \\\n",
    "                '\"cc_num\"              BIGINT,\\n' + \\\n",
    "                '\"num_trans_last_10m\"  SMALLINT,\\n' + \\\n",
    "                '\"avg_amt_last_10m\"    REAL\\n);\\n\\n' + \\\n",
    "            'CREATE OR REPLACE PUMP \"STREAM_PUMP\" AS\\n' + \\\n",
    "            'INSERT INTO \"DESTINATION_SQL_STREAM\"\\n' + \\\n",
    "                'SELECT STREAM \"cc_num\", \\n' + \\\n",
    "                    'COUNT(*) OVER LAST_10_MINUTES, \\n' + \\\n",
    "                    'AVG(\"amount\") OVER LAST_10_MINUTES\\n' + \\\n",
    "                    'FROM \"SOURCE_SQL_STREAM_001\"\\n' + \\\n",
    "                    'WINDOW LAST_10_MINUTES AS (\\n' + \\\n",
    "                        'PARTITION BY \"cc_num\"\\n' + \\\n",
    "                        'RANGE INTERVAL \\'10\\' MINUTE PRECEDING);\\n'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "kda_inputs = [{\n",
    "                'NamePrefix': 'SOURCE_SQL_STREAM',\n",
    "                'KinesisStreamsInput': {\n",
    "                       'ResourceARN': stream_arn,\n",
    "                       'RoleARN': role\n",
    "                },\n",
    "                'InputSchema': {\n",
    "                      'RecordFormat': {\n",
    "                          'RecordFormatType': 'JSON',\n",
    "                          'MappingParameters': {\n",
    "                              'JSONMappingParameters': {\n",
    "                                  'RecordRowPath': '$'\n",
    "                              }\n",
    "                          },\n",
    "                      },\n",
    "                      'RecordEncoding': 'UTF-8',\n",
    "                      'RecordColumns': [\n",
    "                          {'Name': 'cc_num',  'Mapping': '$.cc_num',   'SqlType': 'DECIMAL(1,1)'},\n",
    "                          {'Name': 'merchant','Mapping': '$.merchant', 'SqlType': 'VARCHAR(64)'},\n",
    "                          {'Name': 'amount', 'Mapping': '$.amount', 'SqlType': 'REAL'},\n",
    "                          {'Name': 'zip_code', 'Mapping': '$.zip_code', 'SqlType': 'INTEGER'}\n",
    "                      ]\n",
    "                }\n",
    "              }                         \n",
    "             ]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3> Create Kinesis Data Analytics Application </h3>\n",
    "\n",
    "We first lookup Lambda ARNs from CloudFormation output, then create a Kinesis Data Analytics application that connects its output to the Streaming Lambda. This Lambda will ingest the records and write them to the SageMaker Feature Group."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "kda_outputs = [{'LambdaOutput': {'ResourceARN': lambda_to_fs_arn, 'RoleARN': role},\n",
    "                'Name': 'DESTINATION_SQL_STREAM',\n",
    "                'DestinationSchema': {'RecordFormatType': 'JSON'}}]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'LambdaOutput': {'ResourceARN': 'arn:aws:lambda:us-east-1:835319576252:function:StreamingIngestAggFeatures',\n",
       "   'RoleARN': 'arn:aws:iam::835319576252:role/service-role/AmazonSageMaker-ExecutionRole-20191006T135881'},\n",
       "  'Name': 'DESTINATION_SQL_STREAM',\n",
       "  'DestinationSchema': {'RecordFormatType': 'JSON'}}]"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kda_outputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "ename": "InvalidArgumentException",
     "evalue": "An error occurred (InvalidArgumentException) when calling the CreateApplication operation: Kinesis Analytics service doesn't have sufficient privileges to assume the role: arn:aws:iam::835319576252:role/service-role/AmazonSageMaker-ExecutionRole-20191006T135881. Please check the role provided.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mInvalidArgumentException\u001b[0m                  Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-48-787868d64b40>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      2\u001b[0m                               \u001b[0mInputs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mkda_inputs\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m                               \u001b[0mOutputs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mkda_outputs\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 4\u001b[0;31m                               ApplicationCode=sql_code)\n\u001b[0m",
      "\u001b[0;32m~/anaconda3/envs/python3/lib/python3.6/site-packages/botocore/client.py\u001b[0m in \u001b[0;36m_api_call\u001b[0;34m(self, *args, **kwargs)\u001b[0m\n\u001b[1;32m    355\u001b[0m                     \"%s() only accepts keyword arguments.\" % py_operation_name)\n\u001b[1;32m    356\u001b[0m             \u001b[0;31m# The \"self\" in this scope is referring to the BaseClient.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 357\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_make_api_call\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0moperation_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    358\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    359\u001b[0m         \u001b[0m_api_call\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mstr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpy_operation_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/python3/lib/python3.6/site-packages/botocore/client.py\u001b[0m in \u001b[0;36m_make_api_call\u001b[0;34m(self, operation_name, api_params)\u001b[0m\n\u001b[1;32m    674\u001b[0m             \u001b[0merror_code\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mparsed_response\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Error\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m{\u001b[0m\u001b[0;34m}\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Code\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    675\u001b[0m             \u001b[0merror_class\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexceptions\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfrom_code\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0merror_code\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 676\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0merror_class\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mparsed_response\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0moperation_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    677\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    678\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mparsed_response\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mInvalidArgumentException\u001b[0m: An error occurred (InvalidArgumentException) when calling the CreateApplication operation: Kinesis Analytics service doesn't have sufficient privileges to assume the role: arn:aws:iam::835319576252:role/service-role/AmazonSageMaker-ExecutionRole-20191006T135881. Please check the role provided."
     ]
    }
   ],
   "source": [
    "kda_client.create_application(ApplicationName='cc-agg-app', \n",
    "                              Inputs=kda_inputs,\n",
    "                              Outputs=kda_outputs,\n",
    "                              ApplicationCode=sql_code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kda_client.describe_application(ApplicationName='cc-agg-app')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kda_client.start_application(ApplicationName='cc-agg-app',\n",
    "                             InputConfigurations=[{'Id': '1.1',\n",
    "                                                   'InputStartingPositionConfiguration': \n",
    "                                                     {'InputStartingPosition':'NOW'}}])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
